package gbench.appdemo.myworld.junit.eco;

import gbench.appdemo.myworld.economy.Economy;
import gbench.appdemo.myworld.util.Latch;
import gbench.common.tree.LittleTree.IRecord;
import gbench.common.tree.LittleTree.SQL;
import gbench.commonApp.data.DataApp;

import static gbench.common.tree.LittleTree.IRecord.*;
import static gbench.common.tree.LittleTree.SQL.nsql;
import static gbench.common.tree.LittleTree.*;
import static gbench.common.tree.LittleTree.KVPair.*;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.DoubleSummaryStatistics;
import java.util.LinkedList;
import java.util.List;
import java.util.Optional;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import org.junit.jupiter.api.Test;


/**
 * 构造一个经济体：与JunitE1的区别就是 增加 消费者接收的数据进行持久化的 功能，于是就引入了数据库操作。
 * @author gbench
 *
 */
public class JunitE2 {
    
    /**
     * 发送消息
     * @param eco 经济体
     */
    public void send(final Economy eco) {
        while(true) {
            final var flag = ai.get()>=MAX_COUNT; // 结束标记
            final var proto = REC("stopflag",flag);
            eco.publish(proto.derive("producer",1,"id",ai.getAndIncrement(),
                "name",rnd(names),"amount",rand.nextInt(10000)
            ));
            if(flag) break; // 停止运行信号
        }// while
    }
    
    /**
     * 保存 event 消息事件
     * @param event
     */
    public void persist(final IRecord event) {
       
        es.execute(()->{ // 刷新数据
            
            final var threadLocal = new ThreadLocal<List<SQL>>();// 线程本地存储
            synchronized(sqls) { // 在 SQL上进行同步
                final var data = A(event.get("id"),event.get("name"), event.get("amount"), LocalDateTime.now()); // 数据参数
                sqls.add(nsql(inserttpl,REC("entries",Optional.of(data).map(rb::get).map(IRecord::L).get())));// 加入SQL的缓存
                if( sqls.size()>=batchsize || event.bool("stopflag")) { // 分批插入 的时机 获取
                    threadLocal.set(new ArrayList<>(sqls));// 把SQL复制到本地线程存储以准备数据吃就好
                    sqls.clear(); // 清空 SQL列表缓存
                }//if 批次写入
            }// synchronized
            
            if( threadLocal.get()!=null ) synchronized(dataApp) {
                dataApp.withTransaction( sess -> threadLocal.get().forEach(trycatch2(sess::sqlexecute)) );
            } // synchronized
            
            if(event.bool("stopflag")) latch.countdown(); // 发送停止运行 命令
        }); //  execute
        
    }
    
    /**
     * 消息接收函数
     * @param event
     */
    public void receive(final IRecord event) {
        System.out.println("1)"+event);
        storage.add(event);// 添加入storage
        this.persist(event);
    }
    
    @Test
    public void foo() {
        // 创建数据表
        dataApp.withTransaction(sess->TUPLES(createsqls).filter(t->t._1()!=1).map(t->t._2()).forEach(trycatch2(sess::sqlexecute)));
        
        new Economy(1024) // 构造一个经济体
        .produce(this::send) // 发送消息
        .consume(this::receive) // 接收消息
        .collect(eco -> {
            latch.await();// 等待producer 运行结束
            eco.shutdown();
            es.shutdown();
            System.out.println("经济体 停止运行");
            program_latch.countdown();
        });
        
        program_latch.await();
        
        // 查看数据结果
        dataApp.withTransaction(sess->{
            TUPLES( storage,sess.sql2records(MFT("select * from {0}",tblname)))
            .forEach(JunitE2::stats); // 对name 进行透视
        });
        
        System.out.println("程序运行结束");
    }
    
    /**
     * 数据统计
     * @param tup 数据分组(分组编号,数据流项目)
     */
    public static void stats(final Tuple2<Long,List<IRecord>> tup) {
        System.out.println("-------"+tup._1()+"-------");
        tup._2().stream().collect(pvtclc("name")).dfs_eval_forall2((path,rr)-> KVP ( 
            Arrays.asList(("E2"+path).split("/")), // 构造路径序列
            rr.collect(Collectors.summarizingDouble(e->((IRecord)e).dbl("amount"))) // 路净值
        )).collect( ssclc( compose_f( IRecord::MAP2REC, IRecord::KVPSMERGE2 ) ) ) // 提取路径层级并 展缓恒树形节点
        .treeNode().forEach( node -> { // 转换成树形结构节点。
            final var dss = node.attr("value",DoubleSummaryStatistics.class); // 提取统计结果
            System.out.println(" -- ".repeat(node.getLevel()-1)+ node.getPath()+"\t"+ (dss==null?"-":dss.getSum()+"/"+dss.getCount()) );
        });
    }
    
    // 数据操作类
    final class MyDataApp extends DataApp {
        public MyDataApp(String dbname) { this.reload(this.configureDbName(dbname));}
        public IRecord configureDbName(final String dbname) { return 
            cfg.derive("url", MFT("jdbc:mysql://localhost:3306/{0}?serverTimezone=GMT%2B8", dbname));
        }
    }; // MyDataApp 数据数据库操作类
    
    
    private final Random rand = new Random(); // 随机器生成器
    private final int batchsize = 1000; // 插入数据的批次大小
    private final Integer MAX_COUNT = 100000; // 插入数据总量
    private final List<SQL> sqls = new LinkedList<>(); // SQL 列表
    private final List<IRecord> storage = new LinkedList<>(); // 消息缓冲
    
    private final String[] names = "甲苯,甲醇,乙醇,丁二醇".split(","); // 产品名称
    private final IRecord proto = REC("#id",1,"~code",1,"name","甲苯","amount",2d,"createtime",LocalDateTime.now());// 原型结构
    private final List<String> keys = proto.aoks2rec(compose_f(e->e.str("name"),SQL::parseFieldName)).filter2("id").keys(); // 提取键名列表
    private final String tblname = "t_double_entry"; // 表名
    private final SQL.Builder sb = SQL.builder(tblname); // 狗仔entry sql 系列构建器具
    private final IRecord.Builder rb = IRecord.builder(proto.filter2("#id")); // 去除id
    private final List<String> createsqls = sb.get(proto).createTable(true); // 生成createsqls
    private final String s1 = JOIN( keys, "," ); // 设置键名
    private final String s2 = JOIN( keys.stream().map(e -> "e." + e), "," ); // 键名
    private final String inserttpl = "insert into " + tblname + " (" + s1 + ") values ${ foreach e in entries (" + s2 + ") }";
    
    private final AtomicInteger ai = new AtomicInteger(0); //计数器
    private final Latch program_latch = new Latch(1); // 程序运行 锁
    private final Latch latch = new Latch(1);// 生产活动控制锁, latch 需要用生产者的数量为基础
    private final MyDataApp dataApp = new MyDataApp("db1");
    private final ExecutorService es = Executors.newFixedThreadPool(5);
    
}